package org.dreamwork.cai.io.netty;

import com.google.gson.Gson;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.DatagramPacket;
import io.netty.channel.socket.InternetProtocolFamily;
import io.netty.channel.socket.nio.NioDatagramChannel;
import org.dreamwork.cai.io.AbstractRouteBroadcaster;
import org.dreamwork.cai.core.BroadcastPayload;
import org.dreamwork.cai.proxy.IServiceResolver;
import org.dreamwork.cai.tools.Misc;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.NetworkInterface;
import java.nio.charset.StandardCharsets;
import java.util.List;

public class NettyRouteBroadcaster extends AbstractRouteBroadcaster {
    private final Logger logger = LoggerFactory.getLogger (NettyRouteBroadcaster.class);

    private final EventLoopGroup receiverGroup = new NioEventLoopGroup ();
    private final EventLoopGroup senderGroup   = new NioEventLoopGroup ();

    private static final int DIR_RECEIVER = 0, DIR_SENDER = 1;

    private NioDatagramChannel sender;
    private InetSocketAddress groupAddress;

    public NettyRouteBroadcaster (IServiceResolver resolver, int senderPort) {
        super (resolver, senderPort);
    }

    @Override
    protected void send (BroadcastPayload payload) {
        Gson g = new Gson ();
        String content = g.toJson (payload);
        byte[] buff = content.getBytes (StandardCharsets.UTF_8);
        DatagramPacket packet = new DatagramPacket (Unpooled.wrappedBuffer (buff), groupAddress, sender.localAddress ());
        sender.writeAndFlush (packet);
    }

    @Override
    protected void bind0 (int port) throws IOException {
        InetAddress target = InetAddress.getByName ("225.0.0.1");
        groupAddress = new InetSocketAddress (target, port);

        // initial the receiver
        bootstrap (receiverGroup, port, DIR_RECEIVER);
        // initial the udp sender
        sender = bootstrap (senderGroup, senderPort, DIR_SENDER);
    }

    @Override
    protected void unbind0 () {
        receiverGroup.shutdownGracefully ();
        senderGroup.shutdownGracefully ();
    }

    private NioDatagramChannel bootstrap (EventLoopGroup elp, int port, int direction) throws IOException {
        Bootstrap bootstrap = new Bootstrap ();
        bootstrap.group (elp);
        bootstrap.channelFactory ((ChannelFactory<NioDatagramChannel>) () ->
            new NioDatagramChannel(InternetProtocolFamily.IPv4)
        );
        if (direction == DIR_RECEIVER) {
            bootstrap.localAddress ("0.0.0.0", port);
        } else {
            bootstrap.localAddress (port);
        }
        bootstrap.handler (new ChannelInitializer<NioDatagramChannel> () {
            @Override
            protected void initChannel (NioDatagramChannel ch) {
                ch.pipeline ().addLast (new NettyUDPReceiver (resolver, senderPort));
            }
        });

        ChannelFuture future = bootstrap.bind ();
        NioDatagramChannel channel = (NioDatagramChannel) future.channel ();
        List<NetworkInterface> list = Misc.findAllValidNetworkInterfaces ();
        for (NetworkInterface ni : list) {
            channel.joinGroup (groupAddress, ni);
        }
        future.syncUninterruptibly ();
        return channel;
    }
}